package com.hw

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, DataFrameReader, SparkSession}

object Demo10MaYiSenLin {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 1)
      //开启join
      .config("spark.sql.crossJoin.enabled",true)
      .master("local")
      .appName("mySl")
      .getOrCreate()

    val userDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("user_id string,date_dt string,low_carbon int")
      .load("data/user_low_carbon.txt")

    val plantCarbonDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("plant_id string,plant_name string,plant_carbon int")
      .load("data/plant_carbon.txt")

    /**
     * 蚂蚁森林植物申领统计
     * 问题1：假设2017年1月1日开始记录低碳数据（user_low_carbon），假设2017年10月1日之前满足申领条件的用户都申领了一颗p004-胡杨，
     * 剩余的能量全部用来领取“p002-沙柳” 。
     * 统计在10月1日累计申领“p002-沙柳” 排名前10的用户信息；以及他比后一名多领了几颗沙柳。
     */
    import org.apache.spark.sql.functions._
    import spark.implicits._

    val tmpDF: DataFrame = userDF
      //将数据中/替换成-方便取出
      .select($"user_id", regexp_replace($"date_dt", "/", "-") as "dt", $"low_carbon")
      //过滤时间要求
      .where($"dt".<=("2017-10-1") and ($"dt").>=("2017-1-1"))
      //根据用户分区，获得到2017.10.1之前的总能量
      .groupBy($"user_id")
      .agg(sum($"low_carbon") as "ad")
      //貌似不能子查询
      // .where($"ad".>=(plantCarbonDF.where($"plant_name"==="胡杨").select($"plant_carbon")))
      .withColumn("tmp", expr("1"))

    // 添加中间变量聚合表
    val tmp1: DataFrame = plantCarbonDF.where($"plant_name" === "胡杨")
      .select($"plant_carbon" as "huYang")
      .withColumn("tmp",expr("1"))
    val tmp2: DataFrame = plantCarbonDF.where($"plant_name" === "沙柳")
      .select($"plant_carbon" as "shaLiu")
      .withColumn("tmp",expr("1"))
    //将值聚合在一张表
    val joinDF: DataFrame = tmpDF.join(tmp1, "tmp").join(tmp2, "tmp")

    joinDF
      //过滤出满足申领条件的用户
      .where($"ad">$"huYang")
      .withColumn("count",floor(($"ad"-$"huYang")/$"shaLiu"))
      .withColumn("rank",row_number() over Window.orderBy($"count".desc))
      //过滤出前10名
      .where($"rank".<=(10))
      .withColumn("last_count",lead($"count",1,0) over Window.orderBy($"count".desc))
      //与后一名的差值
      .withColumn("diff",$"count"-$"last_count")

      //整理数据
      .select($"user_id",$"count" as "plant_count",$"diff" as "less_count")
      .show()

  }

}
